package cn.doitedu.kafka.demos;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.Set;

public class ConsumerDemo2手动分配消费分区 {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "doit01:9092,doit02:9092,doit03:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 如果消费者从__consumer_offset中获取不到之前记录的消费偏移量，则通过该参数来决定从什么位置拉取数据
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");  //  earliest|latest|none

        // 客户端id
        props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "c001");

        // 消费者组ID
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g001");


        // 构造消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);


        // 订阅主题,使用手动分配partition的方式: assign
        consumer.assign(Arrays.asList(new TopicPartition("doit27-1", 0)));


        // 拉取消息
        boolean flag = true;
        while (flag) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));

            Set<TopicPartition> partitions = records.partitions();
            for (TopicPartition partition : partitions) {
                List<ConsumerRecord<String, String>> records1 = records.records(partition);
                for (ConsumerRecord<String, String> record : records1) {
                    System.out.println(record.key() + "," + record.value() + "->" + record.topic() + ":" + record.partition() + ":" + record.offset());
                }
            }

        }

        consumer.close();

    }
}
